package bigdata.scala.spark.streaming.ycfx.queue

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

import bigdata.java.framework.spark.util.client.HBaseClient
import bigdata.scala.spark.streaming.ycfx.model.{UserPowerCurveModel, UserPowerModel}

class UserPowerQueue {

  def main(): Unit = {
    // 从kafka获取用户电压数据
    UserPowerCurveModel userPowerCurveModel = kafka取一条数据
    putUserPowerHbase(userPowerCurveModel)
  }

  // 将用户电压写入hbase
  def putUserPowerHbase(userPowerCurveModel : UserPowerCurveModel) = {
    val dataDate:String = LocalDateTime.parse(userPowerCurveModel.dataDate, DateTimeFormatter.ofPattern("yyyyMMdd")).toString // 获取时间
    putUserPowerHbaseBy24Point(dataDate, userPowerCurveModel)
  }

  // 将24个点用户电压写入hbase
  def putUserPowerHbaseBy24Point(dataDate : String, userPowerCurveModel : UserPowerCurveModel): Unit = {
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 1, userPowerCurveModel.u1, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第一个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 5, userPowerCurveModel.u5, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第五个点到hbase
    // 其他同理
  }

  // 将一个点的数据写入hbase
  def putUserPowerHbaseByPoint(dataDate : String, phase : String, point : Int, powerValue : String, meterId : String, meterAssetno : String): Unit = {
    // 通过点数生成rowkey
    var rowkey = genRowkey(dataDate, point)
    // 从hbase获取用户电压
    var userPowerModel = getUserPowerHbase(rowkey)
    // 生成用户电压数据
    genUserPower(userPowerModel, phase, powerValue, meterId, meterAssetno)
    // 将生成新的用户电压写入hbase
    HBaseClient.getInstance().put(MyConstants.tab_user_power, rowkey, userPowerModel)
  }

  // 生成rowkey,point代表是第几个点 u1是第一个点 以此类推
  def genRowkey(dataDate : String, point : Int) = {
    val hourNumber = (point - 1) * 15 / 60; //十五分钟一个点，求出来的就是第几个小时
    var dataDateTime = dataDate + (hourNumber < 10 ? "0" + hourNumber.toString : hourNumber.toString)
    dataDateTime
  }

  // 根据rowkey获取hbase中用户电压
  def getUserPowerHbase(rowkey : String) : UserPowerModel = {
    //如果hbase有则取hbase中的数据
    if (HBaseClient.getInstance().exists(MyConstants.tab_user_power, rowkey)) {
      HBaseClient.getInstance().get(MyConstants.tab_user_power, rowkey)
    }
  }

  // 生成电压数据
  def genUserPower(userPowerModel : UserPowerModel, phase : String, point : Int, powerValue : String, meterId : String, meterAssetno : String): Unit = {
    // 如果不存在用户电压则初始化用户电压
    if (userPowerModel == null) {
      userPowerModel = new UserPowerModel
      userPowerModel.meterId = meterId
      userPowerModel.meterAssetno = meterAssetno
    }
    phase match {
      case "1" => {
        userPowerModel.aphase = powerValue
      }
      case "2" => {
        userPowerModel.bphase = powerValue
      }
      case "3" => {
        userPowerModel.cphase = powerValue
      }
    }
  }
}
